package com.xxxx.streaming

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @program: day0316
 * @author: CoreDao
 * @create: 2021-03-18 11:15
 * */

object WCStreaming extends App {

  //开启本地local[2]
  private val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("nc")
  private val ssc = new StreamingContext(conf, Seconds(5))

  private val sparkcontext: SparkContext = ssc.sparkContext
  sparkcontext.setLogLevel("WARN")

  //定义广播变量
  //黑名单---可设置为外部接收文件,以达到不关闭app的情况下动态改变广播变量
  private val broadcast: Broadcast[List[String]] = sparkcontext.broadcast(List("zs","lisi"))


  private val lines: ReceiverInputDStream[String] = ssc.socketTextStream("D-node01", 9999)

  private val result: DStream[(String, Int)] = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)

  //result.print()

  /**
   * 两种Dstream变成RDD的方式
   */
  /**
   * 第一种
   * foreachRDD
   * 对当前DStream对象中的rdd进行操作
   * 动态改变广播变量DStream的每次操作对应都是一个job
   * 内部需要action算子
   */
/*  result.foreachRDD(rdd=>{
    println("in")
    rdd.filter(x=>{
      x._1.contains("hello")
    }).foreach(println)
  })*/


  /**
   * 第二种
   * transform算子
   * Dstream中rdd的操作
   */
  /*result.transform(rdd=>{
    println("++++++++++++")
    rdd
      .filter(tuple =>{
      val value = broadcast.value
      value.contains(tuple._1)
    })
      .map(x=>(x._2,x._1))
  }).print()*/

  /**
   * Stream特殊内容
   */

  ssc.start()

  ssc.awaitTermination()


  //todo
  //关闭流操作，sparkContext依然存在，可以进行rdd的操作
  ssc.stop(false)

  sparkcontext.parallelize(Array(1,2,3)).foreach(println)

  sparkcontext.stop()

}
